{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "2db38aac-d1ee-4cd2-a362-2cdd98da027f",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "8542eea3-bf48-4ca8-b41e-e87a8fafd73b",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "import pyspark.sql.functions as F\n",
    "import pyspark.sql.types as T\n",
    "from pyspark.sql.utils import AnalysisException"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "3b451aca-f59b-482f-8d3a-e152d1c3cdac",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.master(\"local[*]\").appName(\"spark_and_sql\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "8e4b03d3-5d90-40a0-8fbb-ac3d84e4c2a6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- AtomicNumber: integer (nullable = true)\n",
      " |-- Element: string (nullable = true)\n",
      " |-- Symbol: string (nullable = true)\n",
      " |-- AtomicMass: double (nullable = true)\n",
      " |-- NumberofNeutrons: integer (nullable = true)\n",
      " |-- NumberofProtons: integer (nullable = true)\n",
      " |-- NumberofElectrons: integer (nullable = true)\n",
      " |-- Period: integer (nullable = true)\n",
      " |-- Group: integer (nullable = true)\n",
      " |-- Phase: string (nullable = true)\n",
      " |-- Radioactive: string (nullable = true)\n",
      " |-- Natural: string (nullable = true)\n",
      " |-- Metal: string (nullable = true)\n",
      " |-- Nonmetal: string (nullable = true)\n",
      " |-- Metalloid: string (nullable = true)\n",
      " |-- Type: string (nullable = true)\n",
      " |-- AtomicRadius: double (nullable = true)\n",
      " |-- Electronegativity: double (nullable = true)\n",
      " |-- FirstIonization: double (nullable = true)\n",
      " |-- Density: double (nullable = true)\n",
      " |-- MeltingPoint: double (nullable = true)\n",
      " |-- BoilingPoint: double (nullable = true)\n",
      " |-- NumberOfIsotopes: integer (nullable = true)\n",
      " |-- Discoverer: string (nullable = true)\n",
      " |-- Year: integer (nullable = true)\n",
      " |-- SpecificHeat: double (nullable = true)\n",
      " |-- NumberofShells: integer (nullable = true)\n",
      " |-- NumberofValence: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "csv_path = \"DataAnalysisWithPythonAndPySpark-Data-trunk\\elements\\Periodic_Table_Of_Elements.csv\"\n",
    "elements = spark.read.csv(csv_path,\n",
    "                         header = True,\n",
    "                         inferSchema = True)\n",
    "elements.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "c4e5b8ab-8b58-486f-947f-9d58bb8332bd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+\n",
      "|Period|count|\n",
      "+------+-----+\n",
      "|     6|    1|\n",
      "|     4|    1|\n",
      "+------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "elements.where(\"Phase=='liq'\").groupBy(\"Period\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "886652b9-f406-4169-9139-d6aeaa2f7ea9",
   "metadata": {},
   "outputs": [],
   "source": [
    "elements.createOrReplaceTempView(\"elements\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d24a7905-38ed-46d6-a28d-6a6ce1df2051",
   "metadata": {},
   "source": [
    "使用目录`catalog.listTables`查看存在的列表和视图"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "75d4967c-461e-4b55-b5dd-1df5f1b3309e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+\n",
      "|Period|count(1)|\n",
      "+------+--------+\n",
      "|     6|       1|\n",
      "|     4|       1|\n",
      "+------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT Period,count(*) from elements where Phase='liq' GROUP BY Period\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "efdd683f-7b73-4a0e-baae-c3a30b44fb4b",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "39118d2c-37c2-4abe-876b-440f1905b2f0",
   "metadata": {},
   "outputs": [],
   "source": [
    "folder_path = r\"DataAnalysisWithPythonAndPySpark-Data-trunk\\backblaze\\data_Q4_2019\"\n",
    "csv_path_list = [os.path.join(folder_path,file_path) for file_path in os.listdir(folder_path) if file_path.endswith(\".csv\")]\n",
    "backblaze_2019 = spark.read.csv(csv_path_list[0],\n",
    "                               header=True,\n",
    "                               inferSchema=True)\n",
    "schema = backblaze_2019.schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "ae421529-d366-420e-ac6c-b9a2ee6824e8",
   "metadata": {},
   "outputs": [],
   "source": [
    "for csv_path in csv_path_list[1:]:\n",
    "    backblaze_2019 = backblaze_2019.union(spark.read.csv(csv_path,\n",
    "                                               header=True,\n",
    "                                               schema=schema))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "8c35218e-fdde-46d9-8046-fffcaba4a934",
   "metadata": {},
   "outputs": [],
   "source": [
    "backblaze_2019 = backblaze_2019.select([F.col(x).cast(T.LongType()) if x.startswith(\"smart\") else F.col(x) for x in backblaze_2019.columns ])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "90acda3a-8f9f-4759-a6e8-7791aac05a65",
   "metadata": {},
   "outputs": [],
   "source": [
    "backblaze_2019.createOrReplaceTempView(\"backblaze_stat_2019\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "56708c24-0416-4133-99a2-30147fe66cae",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- date: date (nullable = true)\n",
      " |-- serial_number: string (nullable = true)\n",
      " |-- model: string (nullable = true)\n",
      " |-- capacity_bytes: long (nullable = true)\n",
      " |-- failure: integer (nullable = true)\n",
      " |-- smart_1_normalized: long (nullable = true)\n",
      " |-- smart_1_raw: long (nullable = true)\n",
      " |-- smart_2_normalized: long (nullable = true)\n",
      " |-- smart_2_raw: long (nullable = true)\n",
      " |-- smart_3_normalized: long (nullable = true)\n",
      " |-- smart_3_raw: long (nullable = true)\n",
      " |-- smart_4_normalized: long (nullable = true)\n",
      " |-- smart_4_raw: long (nullable = true)\n",
      " |-- smart_5_normalized: long (nullable = true)\n",
      " |-- smart_5_raw: long (nullable = true)\n",
      " |-- smart_7_normalized: long (nullable = true)\n",
      " |-- smart_7_raw: long (nullable = true)\n",
      " |-- smart_8_normalized: long (nullable = true)\n",
      " |-- smart_8_raw: long (nullable = true)\n",
      " |-- smart_9_normalized: long (nullable = true)\n",
      " |-- smart_9_raw: long (nullable = true)\n",
      " |-- smart_10_normalized: long (nullable = true)\n",
      " |-- smart_10_raw: long (nullable = true)\n",
      " |-- smart_11_normalized: long (nullable = true)\n",
      " |-- smart_11_raw: long (nullable = true)\n",
      " |-- smart_12_normalized: long (nullable = true)\n",
      " |-- smart_12_raw: long (nullable = true)\n",
      " |-- smart_13_normalized: long (nullable = true)\n",
      " |-- smart_13_raw: long (nullable = true)\n",
      " |-- smart_15_normalized: long (nullable = true)\n",
      " |-- smart_15_raw: long (nullable = true)\n",
      " |-- smart_16_normalized: long (nullable = true)\n",
      " |-- smart_16_raw: long (nullable = true)\n",
      " |-- smart_17_normalized: long (nullable = true)\n",
      " |-- smart_17_raw: long (nullable = true)\n",
      " |-- smart_18_normalized: long (nullable = true)\n",
      " |-- smart_18_raw: long (nullable = true)\n",
      " |-- smart_22_normalized: long (nullable = true)\n",
      " |-- smart_22_raw: long (nullable = true)\n",
      " |-- smart_23_normalized: long (nullable = true)\n",
      " |-- smart_23_raw: long (nullable = true)\n",
      " |-- smart_24_normalized: long (nullable = true)\n",
      " |-- smart_24_raw: long (nullable = true)\n",
      " |-- smart_168_normalized: long (nullable = true)\n",
      " |-- smart_168_raw: long (nullable = true)\n",
      " |-- smart_170_normalized: long (nullable = true)\n",
      " |-- smart_170_raw: long (nullable = true)\n",
      " |-- smart_173_normalized: long (nullable = true)\n",
      " |-- smart_173_raw: long (nullable = true)\n",
      " |-- smart_174_normalized: long (nullable = true)\n",
      " |-- smart_174_raw: long (nullable = true)\n",
      " |-- smart_177_normalized: long (nullable = true)\n",
      " |-- smart_177_raw: long (nullable = true)\n",
      " |-- smart_179_normalized: long (nullable = true)\n",
      " |-- smart_179_raw: long (nullable = true)\n",
      " |-- smart_181_normalized: long (nullable = true)\n",
      " |-- smart_181_raw: long (nullable = true)\n",
      " |-- smart_182_normalized: long (nullable = true)\n",
      " |-- smart_182_raw: long (nullable = true)\n",
      " |-- smart_183_normalized: long (nullable = true)\n",
      " |-- smart_183_raw: long (nullable = true)\n",
      " |-- smart_184_normalized: long (nullable = true)\n",
      " |-- smart_184_raw: long (nullable = true)\n",
      " |-- smart_187_normalized: long (nullable = true)\n",
      " |-- smart_187_raw: long (nullable = true)\n",
      " |-- smart_188_normalized: long (nullable = true)\n",
      " |-- smart_188_raw: long (nullable = true)\n",
      " |-- smart_189_normalized: long (nullable = true)\n",
      " |-- smart_189_raw: long (nullable = true)\n",
      " |-- smart_190_normalized: long (nullable = true)\n",
      " |-- smart_190_raw: long (nullable = true)\n",
      " |-- smart_191_normalized: long (nullable = true)\n",
      " |-- smart_191_raw: long (nullable = true)\n",
      " |-- smart_192_normalized: long (nullable = true)\n",
      " |-- smart_192_raw: long (nullable = true)\n",
      " |-- smart_193_normalized: long (nullable = true)\n",
      " |-- smart_193_raw: long (nullable = true)\n",
      " |-- smart_194_normalized: long (nullable = true)\n",
      " |-- smart_194_raw: long (nullable = true)\n",
      " |-- smart_195_normalized: long (nullable = true)\n",
      " |-- smart_195_raw: long (nullable = true)\n",
      " |-- smart_196_normalized: long (nullable = true)\n",
      " |-- smart_196_raw: long (nullable = true)\n",
      " |-- smart_197_normalized: long (nullable = true)\n",
      " |-- smart_197_raw: long (nullable = true)\n",
      " |-- smart_198_normalized: long (nullable = true)\n",
      " |-- smart_198_raw: long (nullable = true)\n",
      " |-- smart_199_normalized: long (nullable = true)\n",
      " |-- smart_199_raw: long (nullable = true)\n",
      " |-- smart_200_normalized: long (nullable = true)\n",
      " |-- smart_200_raw: long (nullable = true)\n",
      " |-- smart_201_normalized: long (nullable = true)\n",
      " |-- smart_201_raw: long (nullable = true)\n",
      " |-- smart_218_normalized: long (nullable = true)\n",
      " |-- smart_218_raw: long (nullable = true)\n",
      " |-- smart_220_normalized: long (nullable = true)\n",
      " |-- smart_220_raw: long (nullable = true)\n",
      " |-- smart_222_normalized: long (nullable = true)\n",
      " |-- smart_222_raw: long (nullable = true)\n",
      " |-- smart_223_normalized: long (nullable = true)\n",
      " |-- smart_223_raw: long (nullable = true)\n",
      " |-- smart_224_normalized: long (nullable = true)\n",
      " |-- smart_224_raw: long (nullable = true)\n",
      " |-- smart_225_normalized: long (nullable = true)\n",
      " |-- smart_225_raw: long (nullable = true)\n",
      " |-- smart_226_normalized: long (nullable = true)\n",
      " |-- smart_226_raw: long (nullable = true)\n",
      " |-- smart_231_normalized: long (nullable = true)\n",
      " |-- smart_231_raw: long (nullable = true)\n",
      " |-- smart_232_normalized: long (nullable = true)\n",
      " |-- smart_232_raw: long (nullable = true)\n",
      " |-- smart_233_normalized: long (nullable = true)\n",
      " |-- smart_233_raw: long (nullable = true)\n",
      " |-- smart_235_normalized: long (nullable = true)\n",
      " |-- smart_235_raw: long (nullable = true)\n",
      " |-- smart_240_normalized: long (nullable = true)\n",
      " |-- smart_240_raw: long (nullable = true)\n",
      " |-- smart_241_normalized: long (nullable = true)\n",
      " |-- smart_241_raw: long (nullable = true)\n",
      " |-- smart_242_normalized: long (nullable = true)\n",
      " |-- smart_242_raw: long (nullable = true)\n",
      " |-- smart_250_normalized: long (nullable = true)\n",
      " |-- smart_250_raw: long (nullable = true)\n",
      " |-- smart_251_normalized: long (nullable = true)\n",
      " |-- smart_251_raw: long (nullable = true)\n",
      " |-- smart_252_normalized: long (nullable = true)\n",
      " |-- smart_252_raw: long (nullable = true)\n",
      " |-- smart_254_normalized: long (nullable = true)\n",
      " |-- smart_254_raw: long (nullable = true)\n",
      " |-- smart_255_normalized: long (nullable = true)\n",
      " |-- smart_255_raw: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "backblaze_2019.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "c1508bbf-2f8a-4ae5-9796-7cff35a692b0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+-------+\n",
      "|               model|              min_GB| max_GB|\n",
      "+--------------------+--------------------+-------+\n",
      "|       ST16000NM001G|             14902.0|14902.0|\n",
      "| TOSHIBA MG07ACA14TA|-9.31322574615478...|13039.0|\n",
      "|       ST12000NM0117|             11176.0|11176.0|\n",
      "|       ST12000NM0007|-9.31322574615478...|11176.0|\n",
      "|HGST HUH721212ALE600|             11176.0|11176.0|\n",
      "+--------------------+--------------------+-------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sql_str = \"\"\"\n",
    "SELECT model,MIN(capacity_bytes)/POW(1024,3) min_GB,MAX(capacity_bytes)/POW(1024,3) max_GB\n",
    "FROM backblaze_stat_2019\n",
    "GROUP BY model\n",
    "ORDER BY max_GB DESC\n",
    "\"\"\"\n",
    "spark.sql(sql_str).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "95daef96-b245-44bc-98b2-93bed32e7d14",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+-----------------+\n",
      "|               model|              min_GB|           max_GB|\n",
      "+--------------------+--------------------+-----------------+\n",
      "| TOSHIBA MG07ACA14TA|-9.31322574615478...|          13039.0|\n",
      "|HGST HMS5C4040BLE640|-9.31322574615478...|3726.023277282715|\n",
      "|HGST HMS5C4040ALE640|-9.31322574615478...|3726.023277282715|\n",
      "|         ST6000DX000|-9.31322574615478...| 5589.02986907959|\n",
      "|         ST8000DM002|-9.31322574615478...|7452.036460876465|\n",
      "+--------------------+--------------------+-----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sql_str = \"\"\"\n",
    "SELECT model,MIN(capacity_bytes)/POW(1024,3) min_GB,MAX(capacity_bytes)/POW(1024,3) max_GB\n",
    "FROM backblaze_stat_2019\n",
    "GROUP BY model\n",
    "ORDER BY min_GB\n",
    "\"\"\"\n",
    "spark.sql(sql_str).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "bfa67aae-bcd7-4a16-aff1-09df15ad9198",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+-----------------+\n",
      "|               model|              min_GB|           max_GB|\n",
      "+--------------------+--------------------+-----------------+\n",
      "|         ST4000DM000|-9.31322574615478...|3726.023277282715|\n",
      "|HGST HMS5C4040BLE640|-9.31322574615478...|3726.023277282715|\n",
      "|       ST12000NM0007|-9.31322574615478...|          11176.0|\n",
      "|         ST6000DX000|-9.31322574615478...| 5589.02986907959|\n",
      "|        ST8000NM0055|-9.31322574615478...|7452.036460876465|\n",
      "+--------------------+--------------------+-----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sql_str = \"\"\"\n",
    "SELECT model,MIN(capacity_bytes)/POW(1024,3) min_GB,MAX(capacity_bytes)/POW(1024,3) max_GB\n",
    "FROM backblaze_stat_2019\n",
    "GROUP BY model\n",
    "HAVING min_GB != max_GB\n",
    "ORDER BY min_GB\n",
    "\"\"\"\n",
    "spark.sql(sql_str).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "a2f1c84f-e180-4f87-981e-bd7841e2f69a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+-----------------+\n",
      "|               model|              min_GB|           max_GB|\n",
      "+--------------------+--------------------+-----------------+\n",
      "|         ST4000DM000|-9.31322574615478...|3726.023277282715|\n",
      "|HGST HMS5C4040BLE640|-9.31322574615478...|3726.023277282715|\n",
      "|       ST12000NM0007|-9.31322574615478...|          11176.0|\n",
      "|         ST6000DX000|-9.31322574615478...| 5589.02986907959|\n",
      "|        ST8000NM0055|-9.31322574615478...|7452.036460876465|\n",
      "+--------------------+--------------------+-----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "backblaze_2019.groupBy(\"model\").agg(\n",
    "    (F.min(F.col(\"capacity_bytes\"))/F.pow(F.lit(1024),3)).alias('min_GB'),\n",
    "    (F.max(F.col(\"capacity_bytes\"))/F.pow(F.lit(1024),3)).alias('max_GB')).where(\"min_GB != max_GB\").orderBy(\"min_GB\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2370f1c4-69d7-452b-8ebe-28298fb7afc4",
   "metadata": {},
   "source": [
    "pyspark没有将having作为方法，而是使用了where"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "67af8fd1-523d-4f7c-95e7-84a084ecd8f9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# sql_str = \"\"\"\n",
    "# CREATE TABLE q11\n",
    "# (model STRING,\n",
    "# capacity_bytes LONG)\n",
    "# USING CSV\n",
    "# OPTIONS (\n",
    "#     'header' 'true',\n",
    "#     'path' 'csv_data/2019-07-01.csv',\n",
    "#     'inferSchema' 'true'\n",
    "# )\n",
    "# \"\"\"\n",
    "# spark.sql(sql_str)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "98faa8ef-6570-4d24-a454-1b8d1bf3db8e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[]"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sql_str = \"\"\"\n",
    "CREATE OR REPLACE TEMP VIEW drive_days\n",
    "AS\n",
    "SELECT model,count(*) AS drive_days\n",
    "FROM backblaze_stat_2019\n",
    "GROUP BY model\n",
    "\"\"\"\n",
    "spark.sql(sql_str)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "30ea95b6-bd21-4bac-8758-1686f5056d95",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[]"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sql_str = \"\"\"\n",
    "CREATE OR REPLACE TEMP VIEW failures\n",
    "AS\n",
    "SELECT model,count(*) AS failures\n",
    "FROM backblaze_stat_2019\n",
    "WHERE failure == 1\n",
    "GROUP BY model\n",
    "\"\"\"\n",
    "spark.sql(sql_str)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "8b33ff8c-8637-4a04-be16-4033c9ff31aa",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------+\n",
      "|               model|failures|\n",
      "+--------------------+--------+\n",
      "|       ST12000NM0007|     364|\n",
      "|          ST500LM030|       6|\n",
      "|HGST HMS5C4040ALE640|       4|\n",
      "|         ST4000DM000|     119|\n",
      "|  TOSHIBA MQ01ABF050|      23|\n",
      "+--------------------+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT * FROM failures\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "fc69181c-3be7-4d62-afcd-db66e8b13718",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----------+--------+\n",
      "|              model|drive_days|failures|\n",
      "+-------------------+----------+--------+\n",
      "|        ST4000DM000|   1757603|     119|\n",
      "|      ST12000NM0007|   3395334|     364|\n",
      "|        ST8000DM005|      2257|    NULL|\n",
      "|TOSHIBA MQ01ABF050M|     36818|      10|\n",
      "|       ST8000NM0055|   1316499|      53|\n",
      "+-------------------+----------+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sql_str = \"\"\"\n",
    "SELECT d.model,d.drive_days,f.failures\n",
    "FROM drive_days AS d\n",
    "LEFT JOIN failures AS f\n",
    "ON d.model == f.model\n",
    "\"\"\"\n",
    "spark.sql(sql_str).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "2d45e26c-d357-4729-8dcf-bacf06c0b406",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----------+--------+--------------------+\n",
      "|              model|drive_days|failures|        failure_rate|\n",
      "+-------------------+----------+--------+--------------------+\n",
      "| TOSHIBA MQ01ABF050|     42565|      23|5.403500528603312E-4|\n",
      "|     WDC WD5000LPVX|     19187|       6|3.127117319018085E-4|\n",
      "|      ST500LM012 HN|     45102|      13|2.882355549643031...|\n",
      "|TOSHIBA MQ01ABF050M|     36818|      10|2.716062795371829E-4|\n",
      "|         ST500LM030|     23025|       6| 2.60586319218241E-4|\n",
      "+-------------------+----------+--------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sql_str = \"\"\"\n",
    "WITH drive_days AS(\n",
    "    SELECT model,count(*) AS drive_days\n",
    "    FROM backblaze_stat_2019\n",
    "    GROUP BY model\n",
    "),failures AS(\n",
    "    SELECT model,count(*) AS failures\n",
    "    FROM backblaze_stat_2019\n",
    "    WHERE failure == 1\n",
    "    GROUP BY model\n",
    ")\n",
    "SELECT d.model,d.drive_days,f.failures,f.failures/d.drive_days failure_rate\n",
    "FROM \n",
    "drive_days AS d\n",
    "INNER JOIN\n",
    "failures AS f\n",
    "ON d.model==f.model\n",
    "ORDER BY failure_rate DESC\n",
    "\"\"\"\n",
    "spark.sql(sql_str).show(5,truncate=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "1e9faec0-aa08-4581-8c96-86d80615ca9b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "37"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "elements.where(F.col(\"Radioactive\").isNotNull()).count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "06ea09f3-26ed-4d11-bde0-7c1f05be5040",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+\n",
      "|count|\n",
      "+-----+\n",
      "|   37|\n",
      "+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT COUNT(*) count FROM ELEMENTS WHERE Radioactive IS NOT NULL\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4fe5dffc-e958-422a-8c0c-5c247f6f6035",
   "metadata": {},
   "source": [
    "- reduce函数原本在python2中也是个内置函数，不过在python3中被移到functools模块中。\n",
    "\n",
    "- reduce函数先从列表（或序列）中取出2个元素执行指定函数，并将输出结果与第3个元素传入函数，输出结果再与第4个元素传入函数，…，以此类推，直到列表每个元素都取完。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "47731d49-3f6f-4dd7-b2f4-dfec36e7c488",
   "metadata": {},
   "outputs": [],
   "source": [
    "full_data = backblaze_2019.selectExpr(\"model\",\"capacity_bytes/pow(1024,3) capacity_GB\",\"date\",\"failure\")\n",
    "drive_days = (\n",
    "    full_data.groupby(\"model\",\"capacity_GB\")\n",
    "    .agg(F.count(\"*\").alias(\"drive_days\"))\n",
    ")\n",
    "failures = (\n",
    "    full_data.where(\"failure==1\")\n",
    "    .groupby(\"model\",\"capacity_GB\")\n",
    "    .agg(F.count(\"*\").alias(\"failures\"))\n",
    ")\n",
    "\n",
    "summarized_data = (\n",
    "    drive_days.join(failures,on=[\"model\",'capacity_GB'],how=\"left\")\n",
    "    .fillna(0,[\"failures\"])\n",
    "    .selectExpr(\"model\",\"capacity_GB\",\"failures/drive_days failure_rate\")\n",
    "    .cache()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "02377165-30fa-4075-80ce-cec4b07fea25",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-----------------+--------------------+\n",
      "|              model|      capacity_GB|        failure_rate|\n",
      "+-------------------+-----------------+--------------------+\n",
      "| TOSHIBA MQ01ABF050|465.7617416381836|5.403500528603312E-4|\n",
      "|     WDC WD5000LPVX|465.7617416381836|3.127117319018085E-4|\n",
      "|      ST500LM012 HN|465.7617416381836|2.882355549643031...|\n",
      "|TOSHIBA MQ01ABF050M|465.7617416381836|2.716062795371829E-4|\n",
      "|         ST500LM030|465.7617416381836| 2.60586319218241E-4|\n",
      "+-------------------+-----------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "summarized_data.selectExpr(\"*\").orderBy(\"failure_rate\",ascending=False).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ceda6ef1-be43-43cb-a3a9-88de849b1e93",
   "metadata": {},
   "source": [
    "exercise 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "6ac901f6-e115-48f0-b563-195d87bda8a8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-----------------+--------------------+\n",
      "|              model|      capacity_GB|        failure_rate|\n",
      "+-------------------+-----------------+--------------------+\n",
      "| TOSHIBA MQ01ABF050|465.7617416381836|5.403500528603312E-4|\n",
      "|     WDC WD5000LPVX|465.7617416381836|3.127117319018085E-4|\n",
      "|      ST500LM012 HN|465.7617416381836|2.882355549643031...|\n",
      "|TOSHIBA MQ01ABF050M|465.7617416381836|2.716062795371829E-4|\n",
      "|         ST500LM030|465.7617416381836| 2.60586319218241E-4|\n",
      "+-------------------+-----------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "answer = (\n",
    "    backblaze_2019\n",
    "    .selectExpr(\"model\",\"capacity_bytes/pow(1024,3) capacity_GB\",\"date\",\"failure\")\n",
    "    .groupby([\"model\",\"capacity_GB\"])\n",
    "    .agg(F.count(\"*\").alias(\"drive_days\"),F.sum(\"failure\").alias(\"failures\"))\n",
    "    .selectExpr(\"model\",\"capacity_GB\",\"failures/drive_days failure_rate\")\n",
    "    .fillna(0,[\"failure_rate\"])\n",
    "    .orderBy(\"failure_rate\",\"capacity_GB\",ascending=[False,True])\n",
    ")\n",
    "answer.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "1235b22c-6fd7-4d0e-acc2-d8fd5ffef1d5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-----------------+--------------------+\n",
      "|              model|      capacity_GB|        failure_rate|\n",
      "+-------------------+-----------------+--------------------+\n",
      "| TOSHIBA MQ01ABF050|465.7617416381836|5.403500528603312E-4|\n",
      "|     WDC WD5000LPVX|465.7617416381836|3.127117319018085E-4|\n",
      "|      ST500LM012 HN|465.7617416381836|2.882355549643031...|\n",
      "|TOSHIBA MQ01ABF050M|465.7617416381836|2.716062795371829E-4|\n",
      "|         ST500LM030|465.7617416381836| 2.60586319218241E-4|\n",
      "+-------------------+-----------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sql_str = \"\"\"\n",
    "WITH summary AS(\n",
    "SELECT model,capacity_bytes/POW(1024,3) capacity_GB,COUNT(date) drive_days,SUM(failure) failures\n",
    "FROM backblaze_stat_2019\n",
    "GROUP BY  model,capacity_bytes\n",
    ")\n",
    "SELECT model,capacity_GB,IFNULL(failures/drive_days,0) failure_rate\n",
    "FROM\n",
    "summary\n",
    "ORDER BY failure_rate DESC,model ASC\n",
    "\"\"\"\n",
    "spark.sql(sql_str).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ac493b2e-6636-4489-b75f-c3fde57473bc",
   "metadata": {},
   "source": [
    "exercies2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "5136f667-b938-4cc3-89c5-0e5ddcadd9a3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------------+\n",
      "|      date|total_capacity_TB|\n",
      "+----------+-----------------+\n",
      "|2019-10-01|884306.1266535893|\n",
      "|2019-11-01|923939.4710115194|\n",
      "|2019-12-01|967612.6495165229|\n",
      "+----------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "answer = (\n",
    "    backblaze_2019\n",
    "    .where(\"dayofmonth(date)=1\")\n",
    "    .selectExpr(\"model\",\"capacity_bytes/pow(1024,4) capacity_TB\",\"date\")\n",
    "    .groupBy('date')\n",
    "    .agg(F.expr(\"sum(capacity_TB) total_capacity_TB\"))\n",
    "    .orderBy('date')\n",
    ")\n",
    "answer.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "9ab53934-642b-4f9e-be84-1675e72924d0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+\n",
      "|  max_date|  min_date|\n",
      "+----------+----------+\n",
      "|2019-12-31|2019-10-01|\n",
      "+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "backblaze_2019.agg(F.max(\"date\").alias(\"max_date\"),F.min(\"date\").alias(\"min_date\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ae46e41d-71f2-4b3f-87b9-bb7ee33e8fa5",
   "metadata": {},
   "source": [
    "exercise3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "e811cccd-eef5-4363-848f-f0f800c2d320",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------+-------------+\n",
      "|              model|           max|          min|\n",
      "+-------------------+--------------+-------------+\n",
      "|        ST4000DM000| 4000787030016|            0|\n",
      "|      ST12000NM0007|12000138625024|            0|\n",
      "|        ST8000DM005| 8001563222016|8001563222016|\n",
      "|TOSHIBA MQ01ABF050M|  500107862016| 500107862016|\n",
      "|       ST8000NM0055| 8001563222016|            0|\n",
      "+-------------------+--------------+-------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    backblaze_2019\n",
    "    .selectExpr(\"serial_number\",\"model\",\"case when capacity_bytes<0 then 0 else capacity_bytes end capacity_bytes\")\n",
    "    .groupby('model')\n",
    "    .agg(F.expr(\"max(capacity_bytes) max\"),F.expr(\"min(capacity_bytes) min\"))\n",
    ").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "cdd70539-c6f5-46f5-a061-b9ca0aa20c0e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------+-------+\n",
      "|               model|capacity_bytes|  count|\n",
      "+--------------------+--------------+-------+\n",
      "|         DELLBOSS VD|  480036847616|   5520|\n",
      "|HGST HDS5C4040ALE630| 4000787030016|   2369|\n",
      "|HGST HMS5C4040ALE640| 4000787030016| 253758|\n",
      "|HGST HMS5C4040ALE640|            -1|     51|\n",
      "|HGST HMS5C4040BLE640| 4000787030016|1168492|\n",
      "|HGST HMS5C4040BLE640|            -1|    247|\n",
      "|HGST HMS5C4040BLE641| 4000787030016|     91|\n",
      "|HGST HUH721010ALE600|10000831348736|   1820|\n",
      "|HGST HUH721212ALE600|12000138625024| 143520|\n",
      "|HGST HUH721212ALN604|12000138625024| 995725|\n",
      "|HGST HUH728080ALE600| 8001563222016|  92073|\n",
      "|HGST HUS726040ALE610| 4000787030016|   2570|\n",
      "|HGST HUS726040ALE610|            -1|      1|\n",
      "+--------------------+--------------+-------+\n",
      "only showing top 13 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "model_capacity_mode_df = (\n",
    "    backblaze_2019\n",
    "    .groupBy(\"model\",\"capacity_bytes\")\n",
    "    .agg(F.count(\"capacity_bytes\").alias(\"count\"))\n",
    "    .orderBy(\"model\",\"count\",ascending=[True,False])\n",
    ")\n",
    "model_capacity_mode_df.show(15)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "f4847d61-9348-4a57-b178-79a0af4447a2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------+-------+\n",
      "|               model|capacity_bytes|  count|\n",
      "+--------------------+--------------+-------+\n",
      "|         DELLBOSS VD|  480036847616|   5520|\n",
      "|HGST HDS5C4040ALE630| 4000787030016|   2369|\n",
      "|HGST HMS5C4040ALE640| 4000787030016| 253758|\n",
      "|HGST HMS5C4040BLE640| 4000787030016|1168492|\n",
      "|HGST HMS5C4040BLE641| 4000787030016|     91|\n",
      "|HGST HUH721010ALE600|10000831348736|   1820|\n",
      "|HGST HUH721212ALE600|12000138625024| 143520|\n",
      "|HGST HUH721212ALN604|12000138625024| 995725|\n",
      "|HGST HUH728080ALE600| 8001563222016|  92073|\n",
      "|HGST HUS726040ALE610| 4000787030016|   2570|\n",
      "|Hitachi HDS5C4040...| 4000787030016|    115|\n",
      "|       ST10000NM0086|10000831348736| 109173|\n",
      "|      ST1000LM024 HN| 1000204886016|     91|\n",
      "|       ST12000NM0007|12000138625024|3394893|\n",
      "|       ST12000NM0008|12000138625024| 321275|\n",
      "+--------------------+--------------+-------+\n",
      "only showing top 15 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "model_capacity_mode_df = (\n",
    "    model_capacity_mode_df\n",
    "    .drop_duplicates([\"model\"])\n",
    "    .orderBy(\"model\")\n",
    ")\n",
    "model_capacity_mode_df.show(15)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "0a4913b1-cd88-4853-b93a-f8306b92c6b8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+--------------+--------------+\n",
      "|               model|      date| serial_number|capacity_bytes|\n",
      "+--------------------+----------+--------------+--------------+\n",
      "|HGST HMS5C4040BLE640|2019-10-01|PL1331LAHERLSH| 4000787030016|\n",
      "|HGST HMS5C4040BLE640|2019-10-01|PL1331LAHERL0H| 4000787030016|\n",
      "|        ST8000NM0055|2019-10-01|      ZA174KR6| 8001563222016|\n",
      "|        ST8000NM0055|2019-10-01|      ZA171E4G| 8001563222016|\n",
      "|HGST HMS5C4040BLE640|2019-10-01|PL1331LAHEN9PH| 4000787030016|\n",
      "|HGST HMS5C4040BLE640|2019-10-01|PL1331LAHEK0BH| 4000787030016|\n",
      "|HGST HMS5C4040BLE640|2019-10-01|PL1331LAHEK0RH| 4000787030016|\n",
      "|        ST8000NM0055|2019-10-01|      ZA172BFR| 8001563222016|\n",
      "|        ST8000NM0055|2019-10-01|      ZA172LVP| 8001563222016|\n",
      "|        ST8000NM0055|2019-10-01|      ZA174QLV| 8001563222016|\n",
      "+--------------------+----------+--------------+--------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "fill_abnormal_capacity_model = (\n",
    "    backblaze_2019\n",
    "    .where(\"capacity_bytes=-1\")\n",
    "    .select(\"model\",\"date\",\"serial_number\")\n",
    "    .join(model_capacity_mode_df.drop(\"count\"),on=\"model\",how=\"inner\")\n",
    ")\n",
    "fill_abnormal_capacity_model.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "ddd5cbb0-9bce-48d7-b3c7-f5562d4486cd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+--------------+--------------+\n",
      "|               model|      date| serial_number|capacity_bytes|\n",
      "+--------------------+----------+--------------+--------------+\n",
      "|         ST4000DM000|2019-10-01|      Z305B2QN| 4000787030016|\n",
      "|       ST12000NM0007|2019-10-01|      ZJV0XJQ4|12000138625024|\n",
      "|       ST12000NM0007|2019-10-01|      ZJV0XJQ3|12000138625024|\n",
      "|       ST12000NM0007|2019-10-01|      ZJV0XJQ0|12000138625024|\n",
      "|HGST HMS5C4040ALE640|2019-10-01|PL1331LAHG1S4H| 4000787030016|\n",
      "|        ST8000NM0055|2019-10-01|      ZA16NQJR| 8001563222016|\n",
      "|       ST12000NM0007|2019-10-01|      ZJV02XWG|12000138625024|\n",
      "|       ST12000NM0007|2019-10-01|      ZJV1CSVX|12000138625024|\n",
      "|       ST12000NM0007|2019-10-01|      ZJV02XWA|12000138625024|\n",
      "|        ST8000NM0055|2019-10-01|      ZA18CEBS| 8001563222016|\n",
      "+--------------------+----------+--------------+--------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "normal_capacity_model = (\n",
    "    backblaze_2019\n",
    "    .where(\"capacity_bytes<>-1\")\n",
    "    .select(\"model\",\"date\",\"serial_number\",\"capacity_bytes\")\n",
    ")\n",
    "answer = normal_capacity_model.union(fill_abnormal_capacity_model)\n",
    "answer.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a7ad04bd-10ad-49db-9393-dc01468c9314",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
